package com.education.realtime.app.dim;

import com.education.realtime.common.Constant;
import com.education.realtime.util.FlinkSinkUtil;
import com.education.realtime.util.JDBCUtil;

import com.education.realtime.app.BaseAppV1 ;
import com.education.realtime.bean.TableProcess;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

import lombok.val;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.Arrays;
import java.util.List;


public class DimApp extends BaseAppV1 {
    public static void main(String[] args) {

        new DimApp().init(2001,"DimApp", Constant.TOPIC_ODS_DB);

    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //1.读取ods_db数据
        SingleOutputStreamOperator<String> etlStream = etlData(stream);//清洗脏数据（不要异常数据和delete数据）
        //2.读取配置数据
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(env);
        // 在HBase中创建表
        checkTable(tpStream);
        //3.将配置数据和ods_db数据connect,过滤出维度数据
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataTpStream = connect(etlStream, tpStream);
        //4.过滤不需要的字段
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = filterUselessColumns(dataTpStream);
        //5.将数据写入hbase
        writeToPhoenix(resultStream);
    }

    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getPhoenixSink());
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> filterUselessColumns(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataTpStream) {
        return  dataTpStream.map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
            @Override
            public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                JSONObject data = t.f0;
                List<String> cs = Arrays.asList(t.f1.getSink_columns().split(","));
                data.keySet().removeIf(key -> !cs.contains(key));
                return t;
            }
        });
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(SingleOutputStreamOperator<String> etlStream,
                                                                                 SingleOutputStreamOperator<TableProcess> tpStream) {
        // 1. 把数据流中的数据解析成jsonObject
        SingleOutputStreamOperator<JSONObject> dataJsonStream = etlStream.map(JSON::parseObject);
        // 2. 把配置做成广播流
        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<>("tpState", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpStateDesc);
        // 3. 数据流和广播流进行connect和处理
        return dataJsonStream
                .connect(tpBcStream)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public void processElement(JSONObject value,
                                               BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>.ReadOnlyContext ctx,
                                               Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        //数据流处理
                        // 1. 从广播状态中读取这个条数据对应的配置信息
                        ReadOnlyBroadcastState<String, TableProcess> tpState = ctx.getBroadcastState(tpStateDesc);
                        // 2. 把数据和配置信息组成一个元组, 放入都后序流中
                        String table = value.getString("table");
                        TableProcess tp = tpState.get(table);
                        // ods_db的数据很多, 有维度, 也有实时
                        if (tp != null) {
                            out.collect(Tuple2.of(value.getJSONObject("data"), tp));
                        }
                    }

                    @Override
                    public void processBroadcastElement(TableProcess tp,
                                                        BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>.Context ctx,
                                                        Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        // 把配置放入到广播状态中
                        BroadcastState<String, TableProcess> tpState = ctx.getBroadcastState(tpStateDesc);
                        tpState.put(tp.getSource_table(), tp);
                    }
                });
    }

    private void checkTable(SingleOutputStreamOperator<TableProcess> tpStream) {
        tpStream.map(tp -> {
            // jdbc操作的流程
            // 1. 获取jdbc连接
            String driver = Constant.PHOENIX_DRIVER;
            String url = Constant.PHOENIX_URL;
            Connection conn = JDBCUtil.getPhoenixConnection(driver,url);
            // 2. 定义一个sql语句
            StringBuilder sql = new StringBuilder();
            sql
                    .append("create table if not exists ")
                    .append(tp.getSink_table())
                    .append("(")
                    .append(tp.getSink_columns().replaceAll("([^,]+)","$1 varchar"))
                    .append(", constraint pk primary key(")
                    .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                    .append(")) ")
                    .append(tp.getSink_extend() == null ? "" : tp.getSink_extend());

            System.out.println("建表语句: "+sql);
            // 3. 通过连接得到一个预处理语句
            PreparedStatement ps = conn.prepareStatement(sql.toString());
            // 5. 执行预处理语句
            ps.execute();
            // 6. 提交
            conn.commit();
            // 7.关闭预处理语句
            ps.close();
            if (conn != null && !conn.isClosed()) {
                conn.close();
            }
            return tp;
        });
    }

    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .scanNewlyAddedTableEnabled(true) // eanbel scan the newly added tables fature
                .databaseList("config") // set captured database
                .tableList("config.table_process") // set captured tables [product, user, address]
                .username("root")
                .password("123321")
                .startupOptions(StartupOptions.initial()) // 第一次启动会同步所有数据, 然后使用bin_log监控变化数据
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

       return env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"mysql-cdc")
                .map(json->{
                    JSONObject obj = JSON.parseObject(json);
                    return obj.getObject("after",TableProcess.class);
                });
    }

    private SingleOutputStreamOperator<String> etlData(DataStreamSource<String> stream) {
        return stream.filter(value -> {
            //通过异常判断数据是否正常
            try {
                JSONObject obj = JSON.parseObject(value);
                // ddl语句不要. insert update
                String type = obj.getString("type");
                val data = obj.getJSONObject("data");

                return ("insert".equals(type)
                        || "update".equals(type)
                        || "bootstrap-insert".equals(type))
                        && data != null;
            } catch (Exception e) {
                System.out.println("json数据格式有问题");
                return false;
            }
        });
    }
}
